package com.mjf.gmall.realtime.dim

import com.alibaba.fastjson.JSON
import com.mjf.gmall.realtime.bean.dim.SpuInfo
import com.mjf.gmall.realtime.util.{MyKafkaUtil, OffsetManager}
import org.apache.hadoop.conf.Configuration
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.phoenix.spark._
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 实时维护商品spu维度表
 *
 * 技术栈：
 *  zookeeper/kafka/redis/hadoop/HBase/phoenix
 *
 * phoenix建表语句：
 *  create table gmall0105_spu_info(id varchar primary key, spu_name varchar);
 *
 * SPU：苹果6（商品聚合信息的最小单位），如手机->苹果手机->苹果6，苹果6就是SPU。
 * SKU：土豪金 16G 苹果6 （商品的不可再分的最小单元）。
 */
object SpuInfoApp {
  def main(args: Array[String]): Unit = {

    // 读取Kafka中商品spu的topic
    // 加载流
    val conf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("SpuInfoApp")
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(5))

    val topic = "ods_spu_info"
    val groupId = "spu_info_group"

    // 从redis读取offset
    val kafkaOffsetMap: Map[TopicPartition, Long] = OffsetManager.getOffset(topic, groupId)

    // 从Kafka消费数据
    var recordInputStream: InputDStream[ConsumerRecord[String, String]] = null
    if (kafkaOffsetMap == null && kafkaOffsetMap.size >0) {
      // 如果redis中已经有了offset信息
      recordInputStream = MyKafkaUtil.getKafkaStream(topic, ssc, kafkaOffsetMap, groupId)
    } else {
      recordInputStream = MyKafkaUtil.getKafkaStream(topic, ssc, groupId)
    }

    //获得偏移结束点
    var offsetRanges: Array[OffsetRange] = Array.empty[OffsetRange]  // 写在算子外面为启动时执行
    val inputGetOffsetDstream: DStream[ConsumerRecord[String, String]] = recordInputStream.transform {
      rdd =>
        // 获得本批次的偏移量的结束位置，用于更新redis中的偏移量
        offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges  // 写在算子里面是周期性执行（driver端）
        rdd
    }


    // 写入到HBase中
    inputGetOffsetDstream.foreachRDD{
      rdd =>
        val spuInfoRDD: RDD[SpuInfo] = rdd.map {
          record =>
            val jsonString: String = record.value()
            val spuInfo: SpuInfo = JSON.parseObject(jsonString, classOf[SpuInfo])
            spuInfo
        }

        spuInfoRDD.saveToPhoenix(
          "GMALL0105_SPU_INFO",
          Seq("ID", "SPU_NAME"),
          new Configuration,
          Some("hadoop102,hadoop103,hadoop104:2181")
        )

        // 数据处理结束，偏移量提交到redis
        OffsetManager.saveOffset(topic, groupId, offsetRanges)
    }

    ssc.start()
    ssc.awaitTermination()

  }
}
